package com.atguigu.day07;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

public class Flink06_IntervalJoin {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        KeyedStream<Tuple2<String, Integer>, Tuple> orderKeyedStream = env
                .fromElements(
                        Tuple2.of("xiaohai", 1),
                        Tuple2.of("xiaohai", 2),
                        Tuple2.of("shasha", 3),
                        Tuple2.of("mengge", 4)
                )
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<Tuple2<String, Integer>>forMonotonousTimestamps()
                                .withTimestampAssigner((value, ts) -> value.f1 * 1000L)
                )
                .keyBy("f0");


        KeyedStream<Tuple3<String, Integer, String>, Tuple> skuKeyedStream = env
                .fromElements(
                        Tuple3.of("xiaohai", 1, "海狗人参丸"),
                        Tuple3.of("xiaohai", 11, "海狗人参丸"),
                        Tuple3.of("shasha", 2, "霸王洗发水"),
                        Tuple3.of("shasha", 13, "生姜生发液"),
                        Tuple3.of("mengge", 13, "尿不湿"),
                        Tuple3.of("nana", 15, "奶粉")
                )
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<Tuple3<String, Integer, String>>forMonotonousTimestamps()
                                .withTimestampAssigner((value, ts) -> value.f1 * 1000L)
                )
                .keyBy("f0");

        //拿两条流做间隔链接 关联用户下单前后10S浏览了哪些商品
        orderKeyedStream.intervalJoin(skuKeyedStream)

                //TODO 下边界的时间是 orderKeyedStream这个流中元素的时间 +lowerBound
                //TODO 上边界的时间是 orderKeyedStream这个流中元素的时间 +upperBound
                //TODO 能关联上的数据范围 时间要符合以下条件
                // orderKeyedStream.timestamp + lowerBound <= skuKeyedStream.timestamp <= orderKeyedStream.timestamp + upperBound
                .between(Time.seconds(-10), Time.seconds(10))
                //不包含下边界
                .lowerBoundExclusive()
                //不包含上边界
                .upperBoundExclusive()
                .process(new ProcessJoinFunction<Tuple2<String, Integer>, Tuple3<String, Integer, String>, String>() {
                    /**
                     * @param left  The left element of the joined pair.
                     * @param right The right element of the joined pair.
                     * @param ctx   A context that allows querying the timestamps of the left, right and joined pair.
                     *              In addition, this context allows to emit elements on a side output.
                     * @param out   The collector to emit resulting elements to.
                     * @throws Exception
                     */
                    @Override
                    public void processElement(Tuple2<String, Integer> left, Tuple3<String, Integer, String> right, ProcessJoinFunction<Tuple2<String, Integer>, Tuple3<String, Integer, String>, String>.Context ctx, Collector<String> out) throws Exception {
                        out.collect(left + "->" + right);
                    }
                }).print();

        env.execute();

    }
}
